package com.doitedu.doit30.hbase.client;

import com.doitedu.doit30.hbase.utils.HbaseUtil;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @Date 22.6.9
 * @Created by HANGGE
 * @Description
 * 1  put数据   Put对象确定一行  向Put对象中可以添加多个单元格
 *              put 'tb_stu' , 'uid001' , 'cf1:name' , 'ly'
 *              Bytes.toBytes(任意的数据类型)  可以转换成字节数组
 *    -- 一个Put对象请求一次服务器
 *  2  批量写入 [多个Put写一次  多个Put请求一次服务器]
 */
public class _04DMLDemo {
    public static void main(String[] args) throws Exception {
        // put数据
        Connection conn = HbaseUtil.getHbaseConnection();
        Table table = HbaseUtil.getTable(conn, "tb_stu");
        deleteData(table);
        conn.close();
        table.close();
    }

    private static void deleteData(Table table) throws IOException {
        // 删除数据
        // 删除一行数据
        Delete delete = new Delete(Bytes.toBytes("linge"));
        // 删除一行中的属性
        // 删除单元格的最新版本数据
        delete.addColumn("cf1".getBytes() , "name".getBytes()) ;
        delete.addColumn("cf1".getBytes() , "age".getBytes()) ;
        delete.addColumn("cf1".getBytes() , "job".getBytes()) ;
        // addColumns 彻底删除单元格(多个版本)
        delete.addColumns("cf1".getBytes() , "name".getBytes()) ;
        // 删除整个列族数据
        delete.addFamily("cf1".getBytes()) ;
        table.delete(delete);
    }

    private static void scanData(Table table) throws IOException {
        /**
         * 扫描表中的数据
         * 返回 多行
         */
        Scan scan = new Scan();
        /**
         * 限制查询的列族
         */
        //  scan.addFamily(Bytes.toBytes("cf2")) ;
        // 参数1 列族   参数2 属性
        // scan.addColumn(Bytes.toBytes("cf1") , Bytes.toBytes("name")) ;
        // 设置返回的起始行数据
        scan.withStartRow(Bytes.toBytes("uid001")) ;  // 包含当前行
        //设置结束行   [ )
        scan.withStopRow(Bytes.toBytes("uid003")) ;

        // 多行数据    遍历行  遍历单元格
        ResultScanner resultScanner = table.getScanner(scan);
        Iterator<Result> iterator = resultScanner.iterator();
        // 遍历行
        while(iterator.hasNext()){
            // 获取当前行
            Result result = iterator.next();
            // 解析行数据
            HbaseUtil.showData(result);
        }
    }

    private static void getData(Table table) throws IOException {
        Get get = new Get("uid001".getBytes());
        // 查询数据  Get
        // Result  行数据  有多个单元格  不确定有几个单元格  本质迭代器(单元格s)
        Result result = table.get(get);
        HbaseUtil.showData(result);
        /*  while(result.advance()){  // 判断是否有下一个单元格
         *//**
         * 获取当前单元格
         *  1 行  2 列族  3  属性  4 值   5 时间戳(版本 , 排序)
         *  Cell中封装的数据底层没有数据类型  都是二进制字节
         *    处理二进制字节  1010111010101001010000101  --> 数据
         *    内部提供了一个解析Cell的工具类  方便 正确
         *//*
            Cell cell = result.current();
            //  使用工具类
            byte[] cfB = CellUtil.cloneFamily(cell);
            byte[] rowB = CellUtil.cloneRow(cell);
            byte[] qualifier = CellUtil.cloneQualifier(cell);
            byte[] value = CellUtil.cloneValue(cell);
            String qualifierStr = new String(qualifier);
            if(!"temperament".equals(qualifierStr)){
                System.out.println("行键: "+ new String(rowB)
                        +"   列族: " + new String(cfB)
                        +"   属性: " + new String(qualifier)
                        +"   值: "  + new String(value)
                );
            }else{
                // 打印结果
                System.out.println("行键: "+ new String(rowB)
                        +"   列族: " + new String(cfB)
                        +"   属性: " + new String(qualifier)
                        // 根据写入的数据类型获取值
                        +"   值: "  + Bytes.toDouble(value)
                );

            }
           *//*
             不能正常解析内容
            byte[] bytes = cell.getFamilyArray();
            String cf = new String(bytes);
            System.out.println(cf);*/


    }

    private static void mutatorData(Connection conn) throws IOException {
        // 获取缓存对象
        BufferedMutator buff = conn.getBufferedMutator(TableName.valueOf("tb_stu"));
        Put put1 = new Put("uid002".getBytes());
        put1.addColumn("cf1".getBytes(), "name".getBytes() , "zss".getBytes());
        put1.addColumn("cf1".getBytes(), "job".getBytes() , "coder".getBytes());

        Put put2 = new Put("uid003".getBytes());
        put2.addColumn("cf1".getBytes(), "name".getBytes() , "lss".getBytes());
        put2.addColumn("cf1".getBytes(), "job".getBytes() , "teacher".getBytes());

        List<Put> puts = new ArrayList<>() ;
        puts.add(put1);
        puts.add(put2);

        /**
         * 执行写操作  是否能写入[是否满足写入的条件]
         * 脚本中flush命令 : 将RS内存中的数据 刷写到HDFS上
         * flush(): 将客户端缓存的数据  请求RS put数据
         * mutate  1 先将Put对象缓存在本地
         *         2 达到阈值 会请求RS写数据 ; 也可以手动的触发flush()
         *         3 请求写数据的条件
         *             1.时间间隔  setWriteBufferPeriodicFlush
         *             2.大小 (字节)  默认不能改
         */
        buff.mutate(puts);  // 将数据放在缓存中
        buff.flush();
    }

    private static void putData(Table stu) throws IOException {
        // 确定行
        Put put = new Put("uid001".getBytes()); // 定位region 找到RS 请求RS 存储内存中  flush到HDFS
        // 列族  属性  值
        put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("temperament") , Bytes.toBytes(99.99));
        put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("name") , Bytes.toBytes("liuyan"));
        stu.put(put);
    }
}
